package com.afdor.rws.event.driven.channel.rocketmq.consumer.topic;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.driven.Serialization;
import com.afdor.rws.event.driven.channel.consumer.EventConsumer;
import com.afdor.rws.event.driven.channel.consumer.handler.EventHandler;
import com.afdor.rws.event.driven.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.channel.consumer.handler.ProxyEventHandler;
import com.afdor.rws.event.driven.channel.consumer.handler.TagFilter;
import com.afdor.rws.event.driven.channel.rocketmq.config.RocketMqEventConfig;
import com.afdor.rws.event.driven.channel.rocketmq.utils.RocketMqUtils;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.entities.Event;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * 基于Rocket MQ实现的事件消费者
 *
 * @author 悭梵
 * @date Created in 2018-12-24 17:52
 */
@Slf4j
@Extension(value = "rocketMqTopic", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventConsumer.class)
public class RocketMqEventConsumer implements EventConsumer, MessageListenerConcurrently, MessageListenerOrderly {

    /**
     * Rocket MQ 消费者
     */
    private volatile DefaultMQPushConsumer consumer;

    /**
     * 线程初始化状态
     */
    private volatile boolean init = Boolean.FALSE;
    /**
     * 线程运行状态
     */
    private volatile boolean running = Boolean.FALSE;
    /**
     * 消费者的名称
     */
    @Getter
    @NonNull
    private String consumerName;
    /**
     * 对应事件源
     */
    @Getter
    @NonNull
    private String eventSourceName;
    /**
     * 配置信息
     */
    @Getter
    @NonNull
    private EventConsumerConfig config;
    /**
     * 事件处理
     */
    @Getter
    @NonNull
    private EventHandler eventHandler;
    /**
     * 消费失败重试策略
     */
    @NonNull
    private EventRetryStrategy eventRetryStrategy;
    /**
     * 序列化方式
     */
    @Getter
    @NonNull
    private Serialization serialization;

    @Getter
    private Set<TagFilter> tagFilterList;

    @Override
    public <C extends EventConsumerConfig> C getConfig() {
        return (C) this.config;
    }

    @Override
    public <T extends EventConsumer, C extends EventConsumerConfig> T setConfig(@NonNull C config) {
        checkStatus();
        this.config = config;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setEventSourceName(@NonNull String eventSourceName) {
        checkStatus();
        this.eventSourceName = eventSourceName;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setConsumerName(String consumerName) {
        checkStatus();
        this.consumerName = consumerName;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setEventHandler(@NonNull EventHandler eventHandler) {
        checkStatus();
        this.eventHandler = new ProxyEventHandler(eventHandler);
        return (T) this;
    }

    @Override
    public EventRetryStrategy getEventRetryStrategy() {
        return eventRetryStrategy;
    }

    @Override
    public <T extends EventConsumer> T setEventRetryStrategy(EventRetryStrategy eventRetryStrategy) {
        checkStatus();
        this.eventRetryStrategy = eventRetryStrategy;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setSerialization(@NonNull Serialization serialization) {
        checkStatus();
        this.serialization = serialization;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T addTagFilter(TagFilter tagFilter) {
        checkStatus();
        if (this.tagFilterList == null) {
            this.tagFilterList = new HashSet<>(8);
        }
        this.tagFilterList.add(tagFilter);
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T init() {
        if (getConfig() == null) {
            throw new RuntimeException("消费者配置信息不能为空");
        }
        if (StringUtils.isBlank(getEventSourceName())) {
            throw new RuntimeException("消费者处理事件源不能为空");
        }
        if (getSerialization() == null) {
            throw new RuntimeException("消费者对应事件序列化处理器不能为空");
        }
        if (getEventHandler() == null) {
            throw new RuntimeException("消费者对应事件处理器不能为空");
        }
        this.init = Boolean.TRUE;
        return (T) this;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        Event event = null;
        EventHandler eventHandler = getEventHandler();
        Serialization serialization = getSerialization();
        for (MessageExt message : messageList) {
            try {
                event = serialization.unserialize(new String(message.getBody()));
                eventHandler.handler(event);
            } catch (Exception e) {
                // 重试5次
                if (message.getReconsumeTimes() >= 5) {
                    // 该条消息可采用其他方式存储与消费
                }
                // 重试
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageList, ConsumeOrderlyContext consumeOrderlyContext) {
        // 设置自动提交
        if (!consumeOrderlyContext.isAutoCommit()) {
            consumeOrderlyContext.setAutoCommit(true);
        }
        try {
            Event event = null;
            for (MessageExt message : messageList) {
                event = getSerialization().unserialize(new String(message.getBody()));
                getEventHandler().handler(event);
            }
        } catch (Exception e) {
            // 重试
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }

    @Override
    public void start() {
        if (this.running) {
            return;
        }
        if (!this.init) {
            init();
        }
        this.running = Boolean.TRUE;

        shutdownRocketMQPushConsumer();

        try {
            RocketMqEventConfig rocketMqEventConfig = getConfig().getEventConfig();
            rocketMqEventConfig.getDefaultTopicQueueNums();

            // TODO 此处注意分片
            this.consumer = new DefaultMQPushConsumer(RocketMqUtils.getTopicSubQueueName(getEventSourceName(), getConsumerName()));
            this.consumer.setNamesrvAddr(rocketMqEventConfig.getNamesrvAddr());
            // 程序第一次启动从消息队列头获取数据,如果非第一次启动，那么按照上次消费的位置继续消费
            this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 可以修改每次消费消息的数量，默认设置是每次消费1条
            this.consumer.setConsumeMessageBatchMaxSize(10);
            // 设置订阅模式
            this.consumer.setMessageModel(MessageModel.CLUSTERING);
            // 订阅指定主题下多有Tag类型的消息,都订阅消息
            this.consumer.subscribe(RocketMqUtils.getTopicName(getEventSourceName()), "*");
            // 注册消费的监听
            if (rocketMqEventConfig.isOrderlyConsumer()) {
                this.consumer.registerMessageListener((MessageListenerOrderly) this);
            } else {
                this.consumer.registerMessageListener((MessageListenerConcurrently) this);
            }
            this.consumer.start();
        } catch (Exception e) {
            stop();
            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            }
            throw new RuntimeException("初始化启动消费者失败", e);
        }
    }

    @Override
    public void stop() {
        this.running = Boolean.FALSE;
        shutdownRocketMQPushConsumer();
    }

    @Override
    public void destroy() {
        stop();
        this.init = Boolean.FALSE;
    }

    /**
     * 关闭Rocket MQ的消费者
     */
    private void shutdownRocketMQPushConsumer() {
        if (this.consumer != null) {
            this.consumer.shutdown();
            this.consumer = null;
        }
    }

    /**
     * 验证是否启动
     */
    private void checkStatus() {
        if (this.running) {
            throw new RuntimeException("消费者启动后，不能进行修改");
        }
    }

}
